package com.rocketmq.provider;

import com.alibaba.fastjson.JSON;
import com.rocketmq.Client;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class NormalAsyncProvider extends Client {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    public static void main(String[] args) throws ClientException, IOException {
        String data = "我是测试数据";
        NormalAsyncProvider provider = new NormalAsyncProvider();
        provider.sendMsg(data);
    }

    private final String topic = "demo-normal";

    public void sendMsg(String data) throws ClientException, IOException {
        ExecutorService sendCallbackExecutor = Executors.newCachedThreadPool();

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        final Producer producer = provider.newProducerBuilder()
                .setClientConfiguration(getConfig())
                .setTopics(topic)
                .build();
        try {
            for (int i = 0; i < 10; i++) {
                final Message message = provider.newMessageBuilder()
                        .setTopic(topic)
                        .setKeys("002")
                        .setBody(JSON.toJSONBytes((data + i)))
                        .build();
                final CompletableFuture<SendReceipt> future = producer.sendAsync(message);

                future.whenCompleteAsync((sendReceipt, throwable) -> {
                    if (null != throwable) {
                        logger.error("Failed to send message", throwable);
                        return;
                    }
                    logger.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
                }, sendCallbackExecutor);
            }
        } catch (Throwable t) {
            logger.error("Failed to send message", t);
        }
        producer.close();
    }
}